package com.lzh.file.redis;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.type.TypeFactory;
import org.apache.commons.compress.utils.Lists;
import org.springframework.data.redis.connection.*;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.TimeoutUtils;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.data.redis.serializer.SerializationException;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Redis 访问封装
 */
@Component
public class RedisOperator {

    private ObjectMapper objectMapper;

    private RedisConnectionFactory connectionFactory;

    public RedisOperator() {
    }

    public RedisOperator(ObjectMapper objectMapper, RedisConnectionFactory connectionFactory) {
        this.objectMapper = objectMapper;
        this.connectionFactory = connectionFactory;
    }

    public Set<String> keys(String pattern) {
        byte[] rawKey = serializeString(pattern);
        Set<byte[]> rawKeys = getConnection().keys(rawKey);
        return deserializeStringSet(rawKeys);
    }

    /**
     * Set {@code value} for {@code key}.
     *
     * @param key   must not be {@literal null}.
     * @param value must not be {@literal null}.
     * @see <a href="https://redis.io/commands/set">Redis Documentation: SET</a>
     */
    public <T> void set(String key, T value) {
        byte[] rawKey = serializeString(key);
        byte[] rawValue = serializeJson(value);
        getConnection().set(rawKey, rawValue);
    }

    /**
     * Set the {@code value} and expiration {@code timeout} for {@code key}.
     *
     * @param key     must not be {@literal null}.
     * @param value   must not be {@literal null}.
     * @param timeout the key expiration timeout.
     * @param unit    must not be {@literal null}.
     * @see <a href="https://redis.io/commands/setex">Redis Documentation: SETEX</a>
     */
    public <T> void set(String key, T value, long timeout, TimeUnit unit) {
        byte[] rawKey = serializeString(key);
        byte[] rawValue = serializeJson(value);
        getConnection().setEx(rawKey, TimeoutUtils.toSeconds(timeout, unit), rawValue);
    }

    /**
     * Set the {@code value} and expiration {@code timeout} for {@code key}.
     *
     * @param key     must not be {@literal null}.
     * @param value   must not be {@literal null}.
     * @param timeout must not be {@literal null}.
     * @throws IllegalArgumentException if either {@code key}, {@code value} or {@code timeout} is not present.
     * @see <a href="https://redis.io/commands/setex">Redis Documentation: SETEX</a>
     * @since 2.1
     */
    public <T> void set(String key, T value, Duration timeout) {
        if (TimeoutUtils.hasMillis(timeout)) {
            set(key, value, timeout.toMillis(), TimeUnit.MILLISECONDS);
        } else {
            set(key, value, timeout.getSeconds(), TimeUnit.SECONDS);
        }
    }

    /**
     * Set {@code key} to hold the string {@code value} if {@code key} is absent.
     *
     * @param key   must not be {@literal null}.
     * @param value must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/setnx">Redis Documentation: SETNX</a>
     */
    public <T> Boolean setIfAbsent(String key, T value) {
        byte[] rawKey = serializeString(key);
        byte[] rawValue = serializeJson(value);
        return getConnection().setNX(rawKey, rawValue);
    }

    /**
     * Set {@code key} to hold the string {@code value} and expiration {@code timeout} if {@code key} is absent.
     *
     * @param key     must not be {@literal null}.
     * @param value   must not be {@literal null}.
     * @param timeout the key expiration timeout.
     * @param unit    must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/set">Redis Documentation: SET</a>
     * @since 2.1
     */
    public <T> Boolean setIfAbsent(String key, T value, long timeout, TimeUnit unit) {
        byte[] rawKey = serializeString(key);
        byte[] rawValue = serializeJson(value);

        Expiration expiration = Expiration.from(timeout, unit);
        return getConnection().set(rawKey, rawValue, expiration, RedisStringCommands.SetOption.ifAbsent());
    }

    /**
     * Set {@code key} to hold the string {@code value} and expiration {@code timeout} if {@code key} is absent.
     *
     * @param key     must not be {@literal null}.
     * @param value   must not be {@literal null}.
     * @param timeout must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     * @throws IllegalArgumentException if either {@code key}, {@code value} or {@code timeout} is not present.
     * @see <a href="https://redis.io/commands/set">Redis Documentation: SET</a>
     * @since 2.1
     */
    public <T> Boolean setIfAbsent(String key, T value, Duration timeout) {

        Assert.notNull(timeout, "Timeout must not be null!");

        if (TimeoutUtils.hasMillis(timeout)) {
            return setIfAbsent(key, value, timeout.toMillis(), TimeUnit.MILLISECONDS);
        }

        return setIfAbsent(key, value, timeout.getSeconds(), TimeUnit.SECONDS);
    }

    /**
     * Set {@code key} to hold the string {@code value} if {@code key} is present.
     *
     * @param key   must not be {@literal null}.
     * @param value must not be {@literal null}.
     * @return command result indicating if the key has been set.
     * @throws IllegalArgumentException if either {@code key} or {@code value} is not present.
     * @see <a href="https://redis.io/commands/set">Redis Documentation: SET</a>
     * @since 2.1
     */
    public <T> Boolean setIfPresent(String key, T value) {
        byte[] rawKey = serializeString(key);
        byte[] rawValue = serializeJson(value);
        return getConnection().set(rawKey, rawValue, Expiration.persistent(), RedisStringCommands.SetOption.ifPresent());
    }

    /**
     * Set {@code key} to hold the string {@code value} and expiration {@code timeout} if {@code key} is present.
     *
     * @param key     must not be {@literal null}.
     * @param value   must not be {@literal null}.
     * @param timeout the key expiration timeout.
     * @param unit    must not be {@literal null}.
     * @return command result indicating if the key has been set.
     * @throws IllegalArgumentException if either {@code key}, {@code value} or {@code timeout} is not present.
     * @see <a href="https://redis.io/commands/set">Redis Documentation: SET</a>
     * @since 2.1
     */
    public <T> Boolean setIfPresent(String key, T value, long timeout, TimeUnit unit) {
        byte[] rawKey = serializeString(key);
        byte[] rawValue = serializeJson(value);

        Expiration expiration = Expiration.from(timeout, unit);
        return getConnection().set(rawKey, rawValue, expiration, RedisStringCommands.SetOption.ifPresent());
    }

    /**
     * Set {@code key} to hold the string {@code value} and expiration {@code timeout} if {@code key} is present.
     *
     * @param key     must not be {@literal null}.
     * @param value   must not be {@literal null}.
     * @param timeout must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     * @throws IllegalArgumentException if either {@code key}, {@code value} or {@code timeout} is not present.
     * @see <a href="https://redis.io/commands/set">Redis Documentation: SET</a>
     * @since 2.1
     */
    public <T> Boolean setIfPresent(String key, T value, Duration timeout) {
        if (TimeoutUtils.hasMillis(timeout)) {
            return setIfPresent(key, value, timeout.toMillis(), TimeUnit.MILLISECONDS);
        }

        return setIfPresent(key, value, timeout.getSeconds(), TimeUnit.SECONDS);
    }

    /**
     * Increment an integer value stored as string value under {@code key} by one.
     *
     * @param key must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/incr">Redis Documentation: INCR</a>
     * @since 2.1
     */
    @Nullable
    public Long increment(String key) {
        final byte[] rawKey = serializeString(key);
        return getConnection().incr(rawKey);
    }

    /**
     * Increment an integer value stored as string value under {@code key} by {@code delta}.
     *
     * @param key   must not be {@literal null}.
     * @param delta
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/incrby">Redis Documentation: INCRBY</a>
     */
    public Long increment(String key, final long delta) {
        final byte[] rawKey = serializeString(key);
        return getConnection().incrBy(rawKey, delta);
    }

    /**
     * Increment a floating point number value stored as string value under {@code key} by {@code delta}.
     *
     * @param key   must not be {@literal null}.
     * @param delta
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/incrbyfloat">Redis Documentation: INCRBYFLOAT</a>
     */
    public Double increment(String key, final double delta) {
        final byte[] rawKey = serializeString(key);
        return getConnection().incrBy(rawKey, delta);
    }

    /**
     * Set multiple keys to multiple values using key-value pairs provided in {@code tuple}.
     *
     * @param map must not be {@literal null}.
     * @see <a href="https://redis.io/commands/mset">Redis Documentation: MSET</a>
     */
    public <T> void multiSet(Map<String, T> map) {

        if (map.isEmpty()) {
            return;
        }

        Map<byte[], byte[]> rawKeys = new LinkedHashMap<>(map.size());

        for (Map.Entry<String, T> entry : map.entrySet()) {
            rawKeys.put(serializeString(entry.getKey()), serializeJson(entry.getValue()));
        }
        getConnection().mSet(rawKeys);
    }

    /**
     * Set multiple keys to multiple values using key-value pairs provided in {@code tuple} only if the provided key does
     * not exist.
     *
     * @param map       must not be {@literal null}.
     * @param {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/msetnx">Redis Documentation: MSETNX</a>
     */
    public <T> Boolean multiSetIfAbsent(Map<String, T> map) {

        if (map.isEmpty()) {
            return true;
        }

        Map<byte[], byte[]> rawKeys = new LinkedHashMap<>(map.size());

        for (Map.Entry<String, T> entry : map.entrySet()) {
            rawKeys.put(serializeString(entry.getKey()), serializeJson(entry.getValue()));
        }

        return getConnection().mSetNX(rawKeys);
    }

    /**
     * Get the value of {@code key}.
     *
     * @param key must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/get">Redis Documentation: GET</a>
     */
    public <T> T get(String key, Class<T> type) {
        byte[] rowKey = serializeString(key);
        byte[] rowValue = getConnection().get(rowKey);
        return deserializeJson(rowValue, type);
    }

    /**
     * Get multiple {@code keys}. Values are returned in the order of the requested keys.
     *
     * @param keys must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/mget">Redis Documentation: MGET</a>
     * @deprecated 使用get(Collection < String > keys, Class < T > clazz)代替
     */
    @Deprecated
    public <T> List<T> multiGet(Collection<String> keys, Class<T> clazz) {
        return get(keys, clazz);
    }

    /**
     * Get multiple {@code keys}. Values are returned in the order of the requested keys.
     *
     * @param keys must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/mget">Redis Documentation: MGET</a>
     */
    public <T> List<T> get(Collection<String> keys, Class<T> clazz) {
        if (keys.isEmpty()) {
            return Collections.emptyList();
        }

        byte[][] rawKeys = serializeString(keys);

        List<byte[]> rawValues = getConnection().mGet(rawKeys);
        return deserializeJson(rawValues, clazz);
    }

    /**
     * Determine if given {@code key} exists.
     *
     * @param key must not be {@literal null}.
     * @return {@literal true} if key exists. {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/exists">Redis Documentation: EXISTS</a>
     */
    public Boolean exists(String key) {
        byte[] rawKey = serializeString(key);
        return getConnection().exists(rawKey);
    }

    /**
     * Determine if multiple {@code keys} exists. Values are returned in the order of the requested keys.
     *
     * @param keys must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/mget">Redis Documentation: MGET</a>
     */
    public List<Boolean> exists(Collection<String> keys) {
        RedisConnection connection = getConnection();
        connection.multi();
        for (String key : keys) {
            byte[] rawKey = serializeString(key);
            connection.exists(rawKey);
        }
        List<?> list = connection.exec();
        return (List<Boolean>) list;
    }

    /**
     * Determine if multiple {@code keys} exists. Values are returned in the order of the requested keys.
     *
     * @param keys must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/mget">Redis Documentation: MGET</a>
     */
    public List<Boolean> exists(String... keys) {
        RedisConnection connection = getConnection();
        connection.multi();
        for (String key : keys) {
            byte[] rawKey = serializeString(key);
            connection.exists(rawKey);
        }
        List<?> list = connection.exec();
        return (List<Boolean>) list;
    }

    /**
     * Count how many of the given {@code keys} exist. Providing the very same {@code key} more than once also counts
     * multiple times.
     *
     * @param keys must not be {@literal null}.
     * @return the number of keys existing among the ones specified as arguments. {@literal null} when used in pipeline /
     * transaction.
     * @since 2.1
     */
    public Long countExists(Collection<String> keys) {
        byte[][] rawKeys = serializeString(keys);
        return getConnection().exists(rawKeys);
    }

    public Long countExists(String... keys) {
        byte[][] rawKeys = serializeString(keys);
        return getConnection().exists(rawKeys);
    }

    /**
     * Delete given {@code keys}.
     *
     * @param keys must not be {@literal null}.
     * @return The number of keys that were removed. {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/del">Redis Documentation: DEL</a>
     */
    public Long delete(String... keys) {
        byte[][] rawKeys = serializeString(keys);
        return getConnection().del(rawKeys);
    }

    /**
     * Delete given {@code keys}.
     *
     * @param keys must not be {@literal null}.
     * @return The number of keys that were removed. {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/del">Redis Documentation: DEL</a>
     */
    public Long delete(Set<String> keys) {
        byte[][] rawKeys = serializeString(keys);
        return getConnection().del(rawKeys);
    }

    /**
     * Delete given hash {@code hashKeys}.
     *
     * @param key      must not be {@literal null}.
     * @param hashKeys must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     */
    public Long hashDelete(String key, String... hashKeys) {

        byte[] rawKey = serializeString(key);
        byte[][] rawHashKeys = serializeString(hashKeys);
        return getConnection().hDel(rawKey, rawHashKeys);
    }

    /**
     * Determine if given hash {@code hashKey} exists.
     *
     * @param key     must not be {@literal null}.
     * @param hashKey must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     */
    public Boolean hashExists(String key, String hashKey) {

        byte[] rawKey = serializeString(key);
        byte[] rawHashKey = serializeString(hashKey);

        return getConnection().hExists(rawKey, rawHashKey);
    }

    /**
     * Get value for given {@code hashKey} from hash at {@code key}.
     *
     * @param key     must not be {@literal null}.
     * @param hashKey must not be {@literal null}.
     * @return {@literal null} when key or hashKey does not exist or used in pipeline / transaction.
     */
    public <T> T hashGet(String key, String hashKey, Class<T> clazz) {
        byte[] rawKey = serializeString(key);
        byte[] rawHashKey = serializeString(hashKey);
        byte[] rawHashValue = getConnection().hGet(rawKey, rawHashKey);
        return deserializeJson(rawHashValue, clazz);
    }

    public <T> List<T> hashGetArray(String key, String hashKey, Class<T> clazz) {
        byte[] rawKey = serializeString(key);
        byte[] rawHashKey = serializeString(hashKey);
        byte[] rawHashValue = getConnection().hGet(rawKey, rawHashKey);
        return deserializeJsonArray(rawHashValue, clazz);
    }

    /**
     * Get values for given {@code hashKeys} from hash at {@code key}.
     *
     * @param key      must not be {@literal null}.
     * @param hashKeys must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     */
    public <T> List<T> hashMultiGet(String key, Collection<String> hashKeys, Class<T> clazz) {
        if (hashKeys.isEmpty()) {
            return Collections.emptyList();
        }

        byte[] rawKey = serializeString(key);
        byte[][] rawHashKeys = new byte[hashKeys.size()][];

        int counter = 0;
        for (String hashKey : hashKeys) {
            rawHashKeys[counter++] = serializeString(hashKey);
        }
        List<byte[]> rawValues = getConnection().hMGet(rawKey, rawHashKeys);
        return deserializeJson(rawValues, clazz);
    }


    /**
     * Get size of hash at {@code key}.
     *
     * @param key must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     */
    public Long hashSize(String key) {
        byte[] rawKey = serializeString(key);
        return getConnection().hLen(rawKey);
    }

    /**
     * Set multiple hash fields to multiple values using data provided in {@code m}.
     *
     * @param key must not be {@literal null}.
     * @param map must not be {@literal null}.
     */
    public void hashPutAll(String key, Map<String, Object> map) {
        if (map.isEmpty()) {
            return;
        }

        byte[] rawKey = serializeString(key);

        Map<byte[], byte[]> hashes = new LinkedHashMap<>(map.size());

        for (Map.Entry<String, Object> entry : map.entrySet()) {
            hashes.put(serializeString(entry.getKey()), serializeJson(entry.getValue()));
        }

        getConnection().hMSet(rawKey, hashes);
    }

    public void hashPutAll(String key, Map<String, Object> map, Duration duration) {
        if (map.isEmpty()) {
            return;
        }

        byte[] rawKey = serializeString(key);

        Map<byte[], byte[]> hashes = new LinkedHashMap<>(map.size());

        for (Map.Entry<String, Object> entry : map.entrySet()) {
            hashes.put(serializeString(entry.getKey()), serializeJson(entry.getValue()));
        }

        getConnection().hMSet(rawKey, hashes);
        if (null != duration) {
            getConnection().expire(rawKey, duration.getSeconds());
        }
    }

    /**
     * Set the {@code value} of a hash {@code hashKey}.
     *
     * @param key     must not be {@literal null}.
     * @param hashKey must not be {@literal null}.
     * @param value
     */
    public <T> void hashPut(String key, String hashKey, T value) {
        byte[] rawKey = serializeString(key);
        byte[] rawHashKey = serializeString(hashKey);
        byte[] rawHashValue = serializeJson(value);

        getConnection().hSet(rawKey, rawHashKey, rawHashValue);
    }


    /**
     * Increment {@code value} of a hash {@code hashKey} by the given {@code value}.
     *
     * @param key     must not be {@literal null}.
     * @param hashKey must not be {@literal null}.
     * @param value
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/hincrby">Redis Documentation: HINCRBY</a>
     */
    public Long hashIncrBy(String key, String hashKey, long value) {
        byte[] rawKey = serializeString(key);
        byte[] rawHashKey = serializeString(hashKey);
        return getConnection().hIncrBy(rawKey, rawHashKey, value);
    }

    /**
     * Set the {@code value} of a hash {@code hashKey} only if {@code hashKey} does not exist.
     *
     * @param key     must not be {@literal null}.
     * @param hashKey must not be {@literal null}.
     * @param value
     * @return {@literal null} when used in pipeline / transaction.
     */
    public <T> Boolean hashPutIfAbsent(String key, String hashKey, T value) {
        byte[] rawKey = serializeString(key);
        byte[] rawHashKey = serializeString(hashKey);
        byte[] rawHashValue = serializeJson(value);
        return getConnection().hSetNX(rawKey, rawHashKey, rawHashValue);
    }

    /**
     * Get entry set (values) of hash at {@code key}.
     *
     * @param key must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     */
    public <T> List<T> hashValues(String key, Class<T> clazz) {
        byte[] rawKey = serializeString(key);
        List<byte[]> rawValues = getConnection().hVals(rawKey);
        return deserializeJson(rawValues, clazz);
    }


    /**
     * Get entire hash stored at {@code key}.
     *
     * @param key must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     */
    public <T> Map<String, T> hashEntries(String key, Class<T> clazz) {
        byte[] rawKey = serializeString(key);
        Map<byte[], byte[]> entries = getConnection().hGetAll(rawKey);
        return deserializeHashMap(entries, clazz);
    }


    /**
     * Set time to live for given {@code key} in seconds.
     *
     * @param key     must not be {@literal null}.
     * @param seconds
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/expire">Redis Documentation: EXPIRE</a>
     */
    public Boolean expire(String key, long seconds) {
        byte[] rawKey = serializeString(key);
        return getConnection().expire(rawKey, seconds);
    }

    /**
     * Set time to live for given {@code key} in seconds.
     *
     * @param key      must not be {@literal null}.
     * @param duration
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/expire">Redis Documentation: EXPIRE</a>
     */
    public Boolean expire(String key, Duration duration) {
        byte[] rawKey = serializeString(key);
        return getConnection().expire(rawKey, duration.getSeconds());
    }

    /**
     * 如果已设置超时时间，则直接返回true
     *
     * @param key
     * @param seconds
     * @return
     */
    public Boolean expireIfAbsent(String key, long seconds) {
        if (ttl(key) > -1) {
            return true;
        }
        return expire(key, seconds);
    }

    /**
     * 如果已设置超时时间，则直接返回true
     *
     * @param key
     * @param duration
     * @return
     */
    public Boolean expireIfAbsent(String key, Duration duration) {
        if (ttl(key) > -1) {
            return true;
        }
        return expire(key, duration);
    }

    /**
     * Get the time to live for {@code key} in seconds.
     *
     * @param key must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/ttl">Redis Documentation: TTL</a>
     */
    public long ttl(String key) {
        byte[] rawKey = serializeString(key);
        return getConnection().ttl(rawKey);
    }

    /**
     * Get elements between {@code begin} and {@code end} from list at {@code key}.
     *
     * @param key   must not be {@literal null}.
     * @param start
     * @param end
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/lrange">Redis Documentation: LRANGE</a>
     */
    public <V> List<V> listRange(String key, long start, long end, Class<V> type) {
        byte[] rawKey = serializeString(key);
        List<byte[]> rawValue = getConnection().lRange(rawKey, start, end);
        return deserializeJson(rawValue, type);
    }

    /**
     * Trim list at {@code key} to elements between {@code start} and {@code end}.
     *
     * @param key   must not be {@literal null}.
     * @param start
     * @param end
     * @see <a href="https://redis.io/commands/ltrim">Redis Documentation: LTRIM</a>
     */
    public void listTrim(String key, long start, long end) {
        byte[] rawKey = serializeString(key);
        getConnection().lTrim(rawKey, start, end);
    }

    /**
     * Get the size of list stored at {@code key}.
     *
     * @param key must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/llen">Redis Documentation: LLEN</a>
     */
    public Long listSize(String key) {
        byte[] rawKey = serializeString(key);
        return getConnection().lLen(rawKey);
    }

    /**
     * Prepend {@code value} to {@code key}.
     *
     * @param key   must not be {@literal null}.
     * @param value
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/lpush">Redis Documentation: LPUSH</a>
     */
    public <V> Long listLeftPush(String key, V value) {
        byte[] rawKey = serializeString(key);
        byte[] rawValue = serializeJson(value);
        return getConnection().lPush(rawKey, rawValue);
    }

    /**
     * Prepend {@code values} to {@code key}.
     *
     * @param key    must not be {@literal null}.
     * @param values
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/lpush">Redis Documentation: LPUSH</a>
     */
    public <V> Long listLeftPushAll(String key, V... values) {
        byte[] rawKey = serializeString(key);
        byte[][] rawValues = serializeJson(values);
        return getConnection().lPush(rawKey, rawValues);
    }

    /**
     * Removes and returns first element in list stored at {@code key}.
     *
     * @param key must not be {@literal null}.
     * @return can be {@literal null}.
     * @see <a href="https://redis.io/commands/lpop">Redis Documentation: LPOP</a>
     */
    public <V> V listLeftPop(String key, Class<V> clazz) {
        byte[] rawKey = serializeString(key);
        byte[] rawValue = getConnection().lPop(rawKey);
        return deserializeJson(rawValue, clazz);
    }

    /**
     * Prepend {@code values} to {@code key}.
     *
     * @param key    must not be {@literal null}.
     * @param values must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/lpush">Redis Documentation: LPUSH</a>
     * @since 1.5
     */
    public <V> Long listLeftPushAll(String key, Collection<V> values) {
        byte[] rawKey = serializeString(key);
        byte[][] rawValues = serializeJson(values);
        return getConnection().lPush(rawKey, rawValues);
    }

    /**
     * Prepend {@code values} to {@code key} only if the list exists.
     *
     * @param key   must not be {@literal null}.
     * @param value
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/lpushx">Redis Documentation: LPUSHX</a>
     */
    public <V> Long listLeftPushIfPresent(String key, V value) {
        byte[] rawKey = serializeString(key);
        byte[] rawValue = serializeJson(value);
        return getConnection().lPushX(rawKey, rawValue);
    }

    /**
     * Insert {@code value} to {@code key} before {@code pivot}.
     *
     * @param key   must not be {@literal null}.
     * @param pivot must not be {@literal null}.
     * @param value
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/linsert">Redis Documentation: LINSERT</a>
     */
    public <V> Long listLeftPush(String key, V pivot, V value) {
        byte[] rawKey = serializeString(key);
        byte[] rawPivot = serializeJson(pivot);
        byte[] rawValue = serializeJson(value);
        return getConnection().lInsert(rawKey, RedisListCommands.Position.BEFORE, rawPivot, rawValue);
    }

    /**
     * Append {@code value} to {@code key}.
     *
     * @param key   must not be {@literal null}.
     * @param value
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/rpush">Redis Documentation: RPUSH</a>
     */
    public <V> Long listRightPush(String key, V value) {
        byte[] rawKey = serializeString(key);
        byte[] rawValue = serializeJson(value);
        return getConnection().rPush(rawKey, rawValue);
    }

    /**
     * Append {@code values} to {@code key}.
     *
     * @param key    must not be {@literal null}.
     * @param values
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/rpush">Redis Documentation: RPUSH</a>
     */
    public <V> Long listRightPushAll(String key, V... values) {
        byte[] rawKey = serializeString(key);
        byte[][] rawValues = serializeJson(values);
        return getConnection().rPush(rawKey, rawValues);
    }

    /**
     * Append {@code values} to {@code key}.
     *
     * @param key    must not be {@literal null}.
     * @param values
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/rpush">Redis Documentation: RPUSH</a>
     * @since 1.5
     */
    public <V> Long listRightPushAll(String key, Collection<V> values) {
        byte[] rawKey = serializeString(key);
        byte[][] rawValues = serializeJson(values);
        return getConnection().rPush(rawKey, rawValues);
    }

    /**
     * Append {@code values} to {@code key} only if the list exists.
     *
     * @param key   must not be {@literal null}.
     * @param value
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/rpushx">Redis Documentation: RPUSHX</a>
     */
    public <V> Long listRightPushIfPresent(String key, V value) {
        byte[] rawKey = serializeString(key);
        byte[] rawValue = serializeJson(value);
        return getConnection().rPushX(rawKey, rawValue);
    }

    /**
     * Insert {@code value} to {@code key} after {@code pivot}.
     *
     * @param key   must not be {@literal null}.
     * @param pivot must not be {@literal null}.
     * @param value
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/linsert">Redis Documentation: LINSERT</a>
     */
    public <V> Long listRightPush(String key, V pivot, V value) {
        byte[] rawKey = serializeString(key);
        byte[] rawPivot = serializeJson(pivot);
        byte[] rawValue = serializeJson(value);
        return getConnection().lInsert(rawKey, RedisListCommands.Position.AFTER, rawPivot, rawValue);
    }

    /**
     * Add given {@code values} to set at {@code key}.
     *
     * @param key    must not be {@literal null}.
     * @param values
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/sadd">Redis Documentation: SADD</a>
     */
    public Long sAdd(String key, Object... values) {
        byte[] rawKey = serializeString(key);
        byte[][] rawValues = serializeJson(values);
        return getConnection().sAdd(rawKey, rawValues);
    }

    /**
     * Remove given {@code values} from set at {@code key} and return the number of removed elements.
     *
     * @param key    must not be {@literal null}.
     * @param values
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/srem">Redis Documentation: SREM</a>
     */
    public Long sRemove(String key, Object... values) {
        byte[] rawKey = serializeString(key);
        byte[][] rawValues = serializeJson(values);
        return getConnection().sRem(rawKey, rawValues);
    }

    /**
     * Check if set at {@code key} contains {@code value}.
     *
     * @param key must not be {@literal null}.
     * @param o
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/sismember">Redis Documentation: SISMEMBER</a>
     */
    public Boolean isSMember(String key, Object o) {
        byte[] rawKey = serializeString(key);
        byte[] rawValue = serializeJson(o);
        return getConnection().sIsMember(rawKey, rawValue);
    }

    /**
     * Check if set at {@code key} contains one or more {@code values}.
     *
     * @param key     must not be {@literal null}.
     * @param objects
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/smismember">Redis Documentation: SMISMEMBER</a>
     * @since 2.6
     */
    public Map<Object, Boolean> isSMember(String key, Object... objects) {
        byte[] rawKey = serializeString(key);
        byte[][] rawValues = serializeJson(objects);
        List<Boolean> result = getConnection().sMIsMember(rawKey, rawValues);

        if (result == null || result.size() != objects.length) {
            return null;
        }

        Map<Object, Boolean> isMember = new LinkedHashMap<>(result.size());

        for (int i = 0; i < objects.length; i++) {
            isMember.put(objects[i], result.get(i));
        }

        return isMember;
    }

    /**
     * Get all elements of set at {@code key}.
     *
     * @param key must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/smembers">Redis Documentation: SMEMBERS</a>
     */
    @Nullable
    public <T> Set<T> sMembers(String key, Class<T> clazz) {
        byte[] rawKey = serializeString(key);
        Set<byte[]> rawValues = getConnection().sMembers(rawKey);

        return deserializeJsonSet(rawValues, clazz);
    }

    /**
     * Get size of set at {@code key}.
     *
     * @param key must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/scard">Redis Documentation: SCARD</a>
     */
    public Long sSize(String key) {
        byte[] rawKey = serializeString(key);
        return getConnection().sCard(rawKey);
    }

    /**
     * 模糊查询
     *
     * @param query 查询参数
     * @return list of keys matched
     */
    public List<String> scan(String query) {
        Cursor<byte[]> cursor = getConnection().scan(ScanOptions.scanOptions().match(query).count(1000).build());

        List<String> result = Lists.newArrayList();
        while (cursor.hasNext()) {
            byte[] rawValue = cursor.next();
            result.add(deserializeString(rawValue));
        }
        return result;
    }

    /**
     * Add {@code value} to a sorted set at {@code key}, or update its {@code score} if it already exists.
     *
     * @param key   must not be {@literal null}.
     * @param value the value.
     * @param score the score.
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/zadd">Redis Documentation: ZADD</a>
     */
    public <V> Boolean zAdd(String key, V value, double score) {

        byte[] rawKey = serializeString(key);
        byte[] rawValue = serializeJson(value);
        return getConnection().zAdd(rawKey, score, rawValue);
    }

    /**
     * Add {@code value} to a sorted set at {@code key} if it does not already exists.
     *
     * @param key   must not be {@literal null}.
     * @param value the value.
     * @param score the score.
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/zadd">Redis Documentation: ZADD NX</a>
     * @since 2.5
     */
    public <V> Boolean zAddIfAbsent(String key, V value, double score) {

        byte[] rawKey = serializeString(key);
        byte[] rawValue = serializeJson(value);
        return getConnection().zAdd(rawKey, score, rawValue, RedisZSetCommands.ZAddArgs.ifNotExists());
    }

    /**
     * Returns the number of elements of the sorted set stored with given {@code key}.
     *
     * @param key
     * @return {@literal null} when used in pipeline / transaction.
     * @see #zCard(String)
     * @see <a href="https://redis.io/commands/zcard">Redis Documentation: ZCARD</a>
     */
    public Long zSize(String key) {
        return zCard(key);
    }

    /**
     * Get the size of sorted set with {@code key}.
     *
     * @param key must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/zcard">Redis Documentation: ZCARD</a>
     * @since 1.3
     */
    public Long zCard(String key) {
        byte[] rawKey = serializeString(key);
        return getConnection().zCard(rawKey);
    }

    /**
     * Get elements between {@code start} and {@code end} from sorted set.
     *
     * @param key   must not be {@literal null}.
     * @param start
     * @param end
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/zrange">Redis Documentation: ZRANGE</a>
     */
    public <V> Set<V> zRange(String key, long start, long end, Class<V> clazz) {

        byte[] rawKey = serializeString(key);
        Set<byte[]> rawValues = getConnection().zRange(rawKey, start, end);

        return deserializeJsonSet(rawValues, clazz);
    }

    /**
     * Remove {@code values} from sorted set. Return number of removed elements.
     *
     * @param key    must not be {@literal null}.
     * @param values must not be {@literal null}.
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/zrem">Redis Documentation: ZREM</a>
     */
    public <V> Long zRemove(String key, V... values) {
        byte[] rawKey = serializeString(key);
        byte[][] rawValues = serializeJson(values);

        return getConnection().zRem(rawKey, rawValues);
    }

    /**
     * Remove elements in range between {@code start} and {@code end} from sorted set with {@code key}.
     *
     * @param key   must not be {@literal null}.
     * @param start
     * @param end
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/zremrangebyrank">Redis Documentation: ZREMRANGEBYRANK</a>
     */
    public Long zRemoveRange(String key, long start, long end) {
        byte[] rawKey = serializeString(key);
        return getConnection().zRemRange(rawKey, start, end);
    }

    /**
     * Increment the score of element with {@code value} in sorted set by {@code increment}.
     *
     * @param key   must not be {@literal null}.
     * @param value the value.
     * @param delta the delta to add. Can be negative.
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/zincrby">Redis Documentation: ZINCRBY</a>
     */
    public <V> Double zIncrScore(String key, V value, double delta) {

        byte[] rawKey = serializeString(key);
        byte[] rawValue = serializeJson(value);
        return getConnection().zIncrBy(rawKey, delta, rawValue);
    }

    /**
     * Get the score of element with {@code value} from sorted set with key {@code key}.
     *
     * @param key must not be {@literal null}.
     * @param o   the value.
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/zscore">Redis Documentation: ZSCORE</a>
     */
    public Double zScore(String key, Object o) {
        byte[] rawKey = serializeString(key);
        byte[] rawValue = serializeJson(o);
        return getConnection().zScore(rawKey, rawValue);
    }

    /**
     * Get elements in range from {@code start} to {@code end} from sorted set ordered from high to low.
     *
     * @param key   must not be {@literal null}.
     * @param start
     * @param end
     * @return {@literal null} when used in pipeline / transaction.
     * @see <a href="https://redis.io/commands/zrevrange">Redis Documentation: ZREVRANGE</a>
     */
    public <V> Set<V> zRevRange(String key, long start, long end, Class<V> clazz) {

        byte[] rawKey = serializeString(key);
        Set<byte[]> rawValues = getConnection().zRevRange(rawKey, start, end);

        return deserializeJsonSet(rawValues, clazz);
    }

    /**
     * Publishes the given message to the given channel.
     *
     * @param channel the channel to publish to, must not be {@literal null}.
     * @param message message to publish
     * @see <a href="https://redis.io/commands/publish">Redis Documentation: PUBLISH</a>
     */
    public void convertAndSend(String channel, Object message) {

        Assert.hasText(channel, "a non-empty channel is required");

        byte[] rawChannel = serializeString(channel);
        byte[] rawMessage = serializeJson(message);

        getConnection().publish(rawChannel, rawMessage);
    }

    protected String deserializeString(byte[] bytes) {
        return (bytes == null ? null : new String(bytes, StandardCharsets.UTF_8));
    }

    protected Set<String> deserializeStringSet(Set<byte[]> bytes) {
        return bytes.stream().map(d -> deserializeString(d)).collect(Collectors.toSet());
    }

    /*
     * (non-Javadoc)
     * @see org.springframework.data.redis.serializer.RedisSerializer#serialize(java.lang.T)
     */
    protected byte[] serializeString(String string) {
        return (string == null ? null : string.getBytes(StandardCharsets.UTF_8));
    }


    /*
     * (non-Javadoc)
     * @see org.springframework.data.redis.core.HashOperations#delete(java.lang.Object, java.lang.Object[])
     */
    protected byte[][] serializeString(String... strings) {

        byte[][] rawStrings = new byte[strings.length][];
        int i = 0;
        for (String string : strings) {
            rawStrings[i++] = serializeString(string);
        }
        return rawStrings;
    }

    protected byte[][] serializeString(Collection<String> strings) {

        byte[][] rawStrings = new byte[strings.size()][];
        int i = 0;
        for (String string : strings) {
            rawStrings[i++] = serializeString(string);
        }
        return rawStrings;
    }


    protected <T> byte[] serializeJson(T object) {
        if (object == null) {
            return new byte[0];
        }
        try {
            return this.objectMapper.writeValueAsBytes(object);
        } catch (Exception ex) {
            throw new SerializationException("Could not write JSON: " + ex.getMessage(), ex);
        }
    }

    protected <T> byte[][] serializeJson(T[] objects) {
        if (objects == null) {
            return new byte[0][];
        }
        try {
            byte[][] bytes = new byte[objects.length][];
            for (int i = 0; i < objects.length; i++) {
                bytes[i] = serializeJson(objects[i]);
            }
            return bytes;
        } catch (Exception ex) {
            throw new SerializationException("Could not write JSON: " + ex.getMessage(), ex);
        }
    }

    protected <T> byte[][] serializeJson(Collection<T> objects) {
        if (objects == null) {
            return new byte[0][];
        }
        try {
            byte[][] bytes = new byte[objects.size()][];
            int i = 0;
            Iterator<T> iterator = objects.iterator();
            while (iterator.hasNext()) {
                bytes[i] = serializeJson(iterator.next());
                i++;
            }
            return bytes;
        } catch (Exception ex) {
            throw new SerializationException("Could not write JSON: " + ex.getMessage(), ex);
        }
    }

    protected <T> Set<T> deserializeJsonSet(Set<byte[]> data, Class<T> type) {
        if (data == null || data.size() == 0) {
            return null;
        }
        try {
            return data.stream().map(d -> deserializeJson(d, type)).collect(Collectors.toSet());
        } catch (Exception ex) {
            throw new SerializationException("Could not read JSON: " + ex.getMessage(), ex);
        }
    }

    protected <T> List<T> deserializeJsonArray(byte[] data, Class<T> type) {
        if (data == null || data.length == 0) {
            return null;
        }
        try {
            return this.objectMapper.readValue(data, getJavaType(List.class, type));
        } catch (Exception ex) {
            throw new SerializationException("Could not read JSON: " + ex.getMessage(), ex);
        }
    }

    protected <T> T deserializeJson(byte[] data, Class<T> type) {
        if (data == null || data.length == 0) {
            return null;
        }
        try {
            return this.objectMapper.readValue(data, 0, data.length, getJavaType(type));
        } catch (Exception ex) {
            throw new SerializationException("Could not read JSON: " + ex.getMessage(), ex);
        }
    }

    protected <T> List<T> deserializeJson(List<byte[]> byteList, Class<T> type) {
        if (byteList == null || byteList.size() == 0) {
            return null;
        }

        List<String> strList = new ArrayList<>(byteList.size());
        byteList.forEach(dataByte -> strList.add(deserializeString(dataByte)));

        try {
            return this.objectMapper.readValue(strList.toString(), getJavaType(List.class, type));
        } catch (JsonProcessingException ex) {
            throw new SerializationException("Could not read JSON: " + ex.getMessage(), ex);
        }
    }

    protected <T> Map<String, T> deserializeHashMap(Map<byte[], byte[]> entries, Class<T> clazz) {
        if (entries == null) {
            return null;
        }

        Map<String, T> map = new LinkedHashMap<>(entries.size());

        for (Map.Entry<byte[], byte[]> entry : entries.entrySet()) {
            map.put(deserializeString(entry.getKey()), deserializeJson(entry.getValue(), clazz));
        }
        return map;
    }

    protected JavaType getJavaType(Class<?> clazz) {
        return TypeFactory.defaultInstance().constructType(clazz);
    }

    protected JavaType getJavaType(Class<? extends Collection> collectClass, Class<?> clazz) {
        return TypeFactory.defaultInstance().constructCollectionType(collectClass, clazz);
    }

    private RedisConnection getConnection() {
        return this.connectionFactory.getConnection();
    }
}
